package com.duobaoyu.dby.supply.biz.service.queue.core;


import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.channel.adapter.enums.DistributionQueueJobStatusEnum;
import com.duobaoyu.dby.supply.biz.enums.LockKeys;
import com.duobaoyu.dby.supply.biz.enums.RedisKeys;
import com.duobaoyu.dby.supply.biz.service.queue.service.AbstractDistributionQueueJobConsumer;
import com.duobaoyu.dby.supply.core.cache.service.RedissonService;
import com.duobaoyu.dby.supply.core.common.exception.CustomAssert;
import com.duobaoyu.dby.supply.core.common.util.DeflaterUtils;
import com.duobaoyu.dby.supply.db.entity.message.DelayMessage;
import com.duobaoyu.dby.supply.db.repository.message.DelayMessageRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * Created by shuming on 2023/9/27 15:56
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class DistributionQueueJobConsumerWork {

	private final RedissonService redissonService;
	private final DelayMessageRepository delayMessageRepository;
	private final List<AbstractDistributionQueueJobConsumer> listeners;

	/**
	 * 消费消息
	 *
	 * @param delayMessage
	 */
	public void consume(DelayMessage delayMessage) {
		DistributionQueueMessage distributionQueueMessage = JSON.parseObject(DeflaterUtils.unzipString(delayMessage.getMessage()),
				DistributionQueueMessage.class);
		if(CollectionUtils.isEmpty(listeners)) {
			log.warn("no consumer handle delay message register ioc");
		}
		if (CollectionUtils.isNotEmpty(listeners)) {
			Optional<AbstractDistributionQueueJobConsumer> jobListener = listeners.stream()
					.filter(listener -> Objects.equals(listener.topicName(),distributionQueueMessage.getTopicName())).findAny();
			CustomAssert.isTrue(jobListener.isPresent(),String.format("no consumer handle delay message, deyMessageId = %s",delayMessage.getId()));
			String lockKey = LockKeys.QUEUE_MESSAGE.formatKey(delayMessage.getId());
			redissonService.tryLock(RedisKeys.LOCK_REDIS.formatKey(lockKey),() -> {
				try {
					this.transferStatusToHanding(delayMessage);
					jobListener.get().consume(distributionQueueMessage);
					this.transferStatusToSuccess(delayMessage);
				} catch (Throwable e) {
					log.warn("consume queue throw exception", e);
					this.transferStatusToFailure(delayMessage);
				}
			});
		}
	}

	private void transferStatusToFailure(DelayMessage delayMessage) {
		try {
			delayMessage.setStatus(DistributionQueueJobStatusEnum.FAILURE.getValue());
			delayMessageRepository.updateStatusById(delayMessage.getId(), delayMessage.getStatus(),true);
		} catch (Exception e) {
			log.warn("transferStatusToFailure throw exception,delayMessage={}", JSON.toJSONString(delayMessage), e);
		}
	}

	private void transferStatusToSuccess(DelayMessage delayMessage) {
		try {
			delayMessage.setStatus(DistributionQueueJobStatusEnum.SUCCESS.getValue());
			delayMessageRepository.updateStatusById(delayMessage.getId(), delayMessage.getStatus(),true);
		} catch (Exception e) {
			log.warn("transferStatusToSuccess throw exception,delayMessage={}", JSON.toJSONString(delayMessage), e);
		}
	}

	private boolean transferStatusToHanding(DelayMessage delayMessage) {
		Integer preStatus = delayMessage.getStatus();
		delayMessage.setStatus(DistributionQueueJobStatusEnum.HANDLING.getValue());
		return delayMessageRepository.updateStatusById(delayMessage.getId(), delayMessage.getStatus(), preStatus,false);
	}

	DelayMessage loadJobMessage(String uqKey) {
		return delayMessageRepository.findByUniqueKey(uqKey);
	}
}
